package com.redis.consumer;

import com.redis.entity.Ticket;
import com.redis.mapper.ConcertMapper;
import com.redis.mapper.TicketMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Service
public class RedisStreamConsumer {
    private final RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ConcertMapper concertMapper;
    @Autowired
    private TicketMapper ticketMapper;
    private static final String STREAM_KEY = "updateTickets";
    private static final String CONSUMER_GROUP = "consumerGroup";
    private static final String CONSUMER_NAME = "consumer1";
    
    public RedisStreamConsumer(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
        // 创建消费者组
        try {
            System.out.println("创建消费者组");
            redisTemplate.opsForStream().createGroup(STREAM_KEY, ReadOffset.from("0-0"), CONSUMER_GROUP);
        } catch (Exception e) {
            System.out.println("消费者组已存在，无需重复创建");
        }
    }
    
    @Scheduled(fixedDelay = 1000) // 每秒轮询一次
    public void consumeMessage() {
        // 从 Stream 中读取消息
        List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream().read(Consumer.from(CONSUMER_GROUP, CONSUMER_NAME), StreamReadOptions.empty().block(Duration.ofSeconds(1)), StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()));
        
        if (messages != null && !messages.isEmpty()) {
            for (MapRecord<String, Object, Object> message : messages) {
                Map<Object, Object> values = message.getValue();
                System.out.println("消费者接收到消息: " + values);
                
                Integer cid = Integer.parseInt(values.get("cid").toString());
                Integer uid = Integer.parseInt(values.get("uid").toString());
                Integer tickets = Integer.parseInt(values.get("tickets").toString());
                
                // 更新剩余票数
                concertMapper.updateTickets(cid, tickets);
                
                // 新增购票记录
                Ticket ticket = new Ticket();
                ticket.setCid(cid);
                ticket.setUid(uid);
                ticket.setPurchaseDate(new Date());
                ticketMapper.add(ticket);
                
                // 确认消息已处理
                redisTemplate.opsForStream().acknowledge(STREAM_KEY, CONSUMER_GROUP, message.getId());
            }
        }
    }
}
